package com.atguigu.flink.chapter07.state;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.ArrayList;

public class Flink01_State_Operator_List_Union {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);
        env.enableCheckpointing(1000); // 开启checkpoint

        env
                .socketTextStream("hadoop162",9999)
                .flatMap(new MyFlatMapFunction())
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static class MyFlatMapFunction implements FlatMapFunction<String, String>, CheckpointedFunction {
        ArrayList<String> words = new ArrayList<>();
        private ListState<String> wordState;

        @Override
        public void flatMap(String value, Collector<String> out) throws Exception {
            String[] data = value.split(" ");

            for (String word : data) {
                words.add(word);
            }

            out.collect(words.toString());

            if (value.contains("x")) {
                throw new RuntimeException("我主动抛异常");
            }
        }

        // 保存状态， 周期的执行
        @Override
        public void snapshotState(FunctionSnapshotContext ctx) throws Exception {
            /*wordState.clear();
            wordState.addAll(words);*/
            // 把数据存储到状态中， 当程序重启的时候，可以自动恢复
            wordState.update(words);
        }

        // 初始化状态： 程序启动的执行执行一次，  一般完成对状态的初始化
        @Override
        public void initializeState(FunctionInitializationContext ctx) throws Exception {
            //ctrl+alt+f
            wordState = ctx.getOperatorStateStore()
                    .getUnionListState(new ListStateDescriptor<String>("wordState", String.class));
            Iterable<String> it = wordState.get();

            // 恢复状态
            for (String word : it) {
                System.out.println("恢复状态：" + word);
                words.add(word);
            }
        }
    }
}
